package com.magenta.maxoptra.gis.distancematrix;

import com.magenta.maxoptra.gis.*;
import com.magenta.maxoptra.gis.commons.Logger;
import com.magenta.maxoptra.gis.commons.Pair;
import com.magenta.maxoptra.gis.datasources.GisDataSource;
import com.magenta.maxoptra.gis.datasources.GisDataSources;
import com.magenta.maxoptra.gis.distancematrix.builder.MatrixBuildTask;
import com.magenta.maxoptra.gis.distancematrix.builder.MatrixBuildTaskListener;
import com.magenta.maxoptra.gis.distancematrix.iterators.AsymmetricIterator;
import com.magenta.maxoptra.gis.service.GeoProviderFacade;
import org.joda.time.LocalTime;

import javax.ejb.EJB;
import javax.ejb.Local;
import javax.ejb.TransactionManagement;
import javax.ejb.TransactionManagementType;
import javax.naming.NamingException;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.magenta.maxoptra.gis.GisParameter.enableBatchRouteRequests;

/**
 * @author Sergey Grachev
 */
@Local(DistanceMatrixService.class)
@TransactionManagement(TransactionManagementType.BEAN)
public class DistanceMatrixServiceImpl implements DistanceMatrixService {
    private static final int UPDATE_CACHE_DELAY = 1 * 60 * 1000; // 1 minute
    private static final int UPDATE_CACHE_DELAY_FOR_HIGH_PRIORITY = 2 * 1000; // 5 seconds
    private static final int UPDATE_EXECUTION_DELAY = 1 * 60 * 1000;
    private static final int UPDATE_MAX_BATCH_SIZE = 50;

    @EJB
    private SyncGeoRoutingService geoRoutingService;

    @EJB
    private GeoProviderFacade geoProviderFacade;

    private final Logger logger = new Logger(getClass());
    private final Map<Long, GroupContext> groups = new HashMap<Long, GroupContext>();
    private final BlockingQueue<Command> commandsQueue = new LinkedBlockingQueue<Command>();
    private final Map<Long, List<MatrixBuildTaskGroup>> currentTaskGroups = new ConcurrentHashMap<Long, List<MatrixBuildTaskGroup>>();
    private final Object lockUpdateTask = new Object();
    private final AtomicBoolean scheduleUpdate = new AtomicBoolean(false);
    private final EventsProducer eventsProducer = new EventsProducer();

    private DistanceMatrixDAO db;
    private MatrixBuildTask<MatrixBuildTaskGroup> buildTask;
    private Thread updateThread;
    private Thread commandsThread;

    private final DistanceMatrixServiceProxy serviceProxy = new DistanceMatrixServiceProxy() {
        @Override
        public void stopGroupUpdating(final long groupId) {
            synchronized (lockUpdateTask) {
                if (buildTask != null) {
                    final List<MatrixBuildTaskGroup> list = currentTaskGroups.get(groupId);
                    if (list != null) {
                        for (final MatrixBuildTaskGroup group : list) {
                            buildTask.stopGroupUpdating(group);
                        }
                    }
                }
            }
        }

        @Override
        public void changeRequestsDelay(final long groupId, final int delay) {
            synchronized (lockUpdateTask) {
                if (buildTask != null) {
                    final List<MatrixBuildTaskGroup> list = currentTaskGroups.get(groupId);
                    if (list != null) {
                        for (final MatrixBuildTaskGroup group : list) {
                            buildTask.changeRequestsDelay(group, delay);
                        }
                    }
                }
            }
        }

        @Override
        public void changeMatrixPriority(final long groupId, final long matrixId, final long priority) {
            synchronized (lockUpdateTask) {
                if (buildTask != null) {
                    final List<MatrixBuildTaskGroup> list = currentTaskGroups.get(groupId);
                    if (list != null) { // find in current task
                        for (final MatrixBuildTaskGroup group : list) {
                            if (group.getMatrixId() == matrixId) {
                                buildTask.changeMatrixPriority(group, priority);
                                return;
                            }
                        }
                    }

                    if (priority == 0) {
                        return;
                    }

                    // add matrix into current task

                    final MatrixBuildTaskGroup matrix = new MatrixBuildTaskGroup(groupId, matrixId, priority);
                    List<MatrixBuildTaskGroup> matrices = currentTaskGroups.get(groupId);
                    if (matrices == null) {
                        currentTaskGroups.put(groupId, matrices = new ArrayList<MatrixBuildTaskGroup>(1));
                    }
                    matrices.add(matrix);

                    buildTask.addGroup(matrix);
                }
            }
        }

        @Override
        public void notifyMatrixUpdateFinished(final long groupId, final long matrixId, final UpdateResultType resultType) {
            eventsProducer.notifyMatrixUpdateFinished(groupId, matrixId, resultType);
        }

        @Override
        public void subscribe(final String queueName, final Set<EventType> events) {
            eventsProducer.subscribe(queueName, events);
        }

        @Override
        public void unsubscribe(final String queueName, final Set<EventType> events) {
            eventsProducer.unsubscribe(queueName, events);
        }
    };

    private final MatrixBuildTaskListener<MatrixBuildTaskGroup> matrixBuildTaskListener = new MatrixBuildTaskListener<MatrixBuildTaskGroup>() {
        @Override
        public MatrixBuildTask.Settings onFetchSettings(final MatrixBuildTaskGroup groupId) {
            final GeoProviderSettings settings = ensureExists(groupId.getGroupId()).getGeoProviderSettings();
            if (settings == null) {
                return null;
            }
            final String providerName = settings.getProviderName().toLowerCase();
            final boolean allowBatchUpdates = Boolean.valueOf(settings.getGisParam(enableBatchRouteRequests.withProvider(providerName)));
            return new MatrixBuildTask.Settings(settings)
                    .withAllowBatchUpdates(allowBatchUpdates)
                    .withCoordinatesLimitPerBatchUpdate(UPDATE_MAX_BATCH_SIZE)
                    .withPriority(groupId.getPriority());
        }

        @Override
        public RoutesIterator onFetchCoordinates(final MatrixBuildTaskGroup groupId) {
            final Pair<Set<DistanceMatrixCoordinate>, Set<DistanceMatrixCoordinate>> coordinates
                    = db.getNewAndOldCoordinates(groupId.getGroupId(), groupId.getMatrixId());
            final Set<DistanceMatrixCoordinate> from = coordinates.getFirst();
            final Set<DistanceMatrixCoordinate> to = coordinates.getSecond();
            final Set<DistanceMatrixCoordinate> all = new HashSet<DistanceMatrixCoordinate>(from.size() + to.size());
            all.addAll(from);
            all.addAll(to);
            return new AsymmetricIterator(new ArrayList<DistanceMatrixCoordinate>(from), new ArrayList<DistanceMatrixCoordinate>(all));
        }

        @Override
        public GeoProviderStatus onUpdateRoute(final GeoProviderSettings settings,
                                               final Pair<? extends GeographicCoordinates, ? extends GeographicCoordinates> route,
                                               final boolean highPriority) {
            final RoutingCoordinate from = new RoutingCoordinate(route.getFirst().getLatitude(), route.getFirst().getLongitude());
            final RoutingCoordinate to = new RoutingCoordinate(route.getSecond().getLatitude(), route.getSecond().getLongitude());
            return geoRoutingService.updateCache(settings, from, to,
                    highPriority ? UPDATE_CACHE_DELAY_FOR_HIGH_PRIORITY : UPDATE_CACHE_DELAY);
        }

        @Override
        public GeoProviderStatus onUpdateRoute(final GeoProviderSettings settings, final GeographicCoordinates from,
                                               final List<? extends GeographicCoordinates> to, final boolean highPriority) {
            return geoRoutingService.updateCache(settings, from, to,
                    highPriority ? UPDATE_CACHE_DELAY_FOR_HIGH_PRIORITY : UPDATE_CACHE_DELAY);
        }

        @Override
        public boolean onCheckIsGroupActive(final MatrixBuildTaskGroup groupId) {
            final GroupContext group = ensureExists(groupId.getGroupId());
            if (!group.isAllowUpdating()) {
                return false;
            }
            final Pair<LocalTime, LocalTime> time = group.getTimeInterval();
            return time == null || checkTime(time);
        }

        @Override
        public void onFinishForCoordinate(final MatrixBuildTaskGroup groupId, final GeographicCoordinates coordinate,
                                          final RoutesIterator.Type routeType, final long providerCid, final boolean hasErrors) {
            if (!hasErrors) {
                db.updateState(groupId.getGroupId(), groupId.getMatrixId(), coordinate,
                        RoutesIterator.Type.FORWARD.equals(routeType), UpdatingState.DONE.getValue(), providerCid);
            }
        }

        @Override
        public void onFinishGroup(final MatrixBuildTaskGroup groupId, final UpdateResultType resultType) {
            eventsProducer.notifyMatrixUpdateFinished(groupId.getGroupId(), groupId.getMatrixId(), resultType);
            db.setMatrixPriority(groupId.getGroupId(), groupId.getMatrixId(), 0);
        }

        @Override
        public void onProgress(final MatrixBuildTaskGroup groupId, final long total, final long progress) {
            eventsProducer.notifyMatrixUpdateProgress(groupId.getGroupId(), groupId.getMatrixId(), total, progress);
        }
    };

    private boolean isDBAvailable() {
        return db != null;
    }

    private GroupContext ensureExists(final long groupId) {
        synchronized (groups) {
            GroupContext groupContext = groups.get(groupId);
            if (groupContext == null) {
                groups.put(groupId, groupContext = new GroupContext(groupId));
            }
            return groupContext;
        }
    }

    private void unlockGroups(final Set<Long> lockedGroups) {
        for (final Long groupId : lockedGroups) {
            final GroupContext groupContext = ensureExists(groupId);
            logger.trace("Unlock group " + groupContext);
            groupContext.unlock();
        }
        lockedGroups.clear();
    }

    private void executeCommands() throws InterruptedException {

        final Set<Long> lockedGroups = new HashSet<Long>();
        long previousGroupId = Long.MIN_VALUE;
        UpdateRequestType requireUpdate = UpdateRequestType.NONE;

        while (Thread.currentThread().isAlive()) {

            boolean successfullyExecuted = false;
            final Command command = commandsQueue.take();

            if (!isDBAvailable()) {
                logger.warn("Database not available, command will be ignored " + command);
                continue;
            }

            if (command instanceof GroupCommand) {

                final GroupCommand groupCommand = (GroupCommand) command;

                try {
                    successfullyExecuted = executeGroupCommand(lockedGroups, previousGroupId, groupCommand);
                } catch (Exception e) {
                    logger.error("Error while execute command: " + command, e);
                }

                requireUpdate = UpdateRequestType.max(requireUpdate, groupCommand.isRequireMatrixUpdating());
                previousGroupId = groupCommand.getGroupId();

            } else {

                unlockGroups(lockedGroups);

                try {
                    successfullyExecuted = executeCommand(command);
                } catch (Exception e) {
                    logger.error("Error while execute command: " + command, e);
                }

                requireUpdate = UpdateRequestType.max(requireUpdate, command.isRequireMatrixUpdating());
                previousGroupId = Long.MIN_VALUE;

            }

            if (!successfullyExecuted) {
                // TODO SG several attempts to run the command
            }

            if (commandsQueue.isEmpty()) {
                unlockGroups(lockedGroups);
                if (!UpdateRequestType.NONE.equals(requireUpdate)) {
                    scheduleUpdate.set(true);
                    if (UpdateRequestType.IMMEDIATELY.equals(requireUpdate)) {
                        synchronized (scheduleUpdate) {
                            scheduleUpdate.notifyAll();
                        }
                    }
                }
            }
        }
    }

    private boolean executeCommand(final Command command) {

        logger.trace("Process command " + command);
        command.execute(db, serviceProxy);

        return true;
    }

    private boolean executeGroupCommand(final Set<Long> lockedGroups, final long previousGroupId, final GroupCommand groupCommand) {

        final GroupContext groupContext = ensureExists(groupCommand.getGroupId());

        // group changed
        if (groupContext.getId() != previousGroupId) {
            // unlock previous
            if (lockedGroups.contains(previousGroupId)) {
                unlockGroups(lockedGroups);
            }

            // lock current
            if (groupCommand.isRequireGroupLock()) {
                try {
                    groupContext.lock();
                    logger.trace("Lock group " + groupContext);
                } finally {
                    lockedGroups.add(groupContext.getId());
                }
            }
        }

        logger.trace("Process group command " + groupCommand);
        groupCommand.useGroupContext(groupContext);
        groupCommand.execute(db, serviceProxy);

        return true;
    }

    private boolean checkTime(final Pair<LocalTime, LocalTime> time) {
        final LocalTime now = new LocalTime();
        return now.isAfter(time.getFirst()) && now.isBefore(time.getSecond());
    }

    private void executeUpdateTask() {
        if (!isDBAvailable()) {
            logger.warn("Database not available, updating skipped");
            return;
        }

        logger.trace("Execute updating task ...");

        currentTaskGroups.clear();

        final Set<MatrixBuildTaskGroup> taskGroups = new HashSet<MatrixBuildTaskGroup>(groups.size());
        final List<Long[]> activeGroups = db.getGroupsForUpdating();
        for (final Long[] activeGroup : activeGroups) {
            final long groupId = activeGroup[0];
            final long matrixId = activeGroup[1];
            final long priority = activeGroup[2];

            final GroupContext groupContext = groups.get(groupId);

            // matrices with priority > 0 updated always
            if (groupContext == null || (priority == 0 && groupContext.isDisabled())) {
                continue;
            }

            // to early or to late for updating (matrices with priority > 0 updated always)
            final Pair<LocalTime, LocalTime> timeInterval = groupContext.getTimeInterval();
            if (priority == 0 && timeInterval != null && !checkTime(timeInterval)) {
                continue;
            }

            final MatrixBuildTaskGroup group = new MatrixBuildTaskGroup(groupId, matrixId, priority);
            if (taskGroups.add(group)) {
                List<MatrixBuildTaskGroup> list = currentTaskGroups.get(group.getGroupId());
                if (list == null) {
                    currentTaskGroups.put(group.getGroupId(), list = new ArrayList<MatrixBuildTaskGroup>(1));
                }
                list.add(group);
            }
        }

        // nothing update
        if (taskGroups.isEmpty()) {
            return;
        }

        synchronized (lockUpdateTask) {
            buildTask = MatrixBuildTask.newInstance(new ArrayList<MatrixBuildTaskGroup>(taskGroups), matrixBuildTaskListener);
        }

        try {
            buildTask.run();
        } catch (Exception e) {
            logger.error("Error while update distance matrices", e);
            scheduleUpdate.compareAndSet(false, true);
        }

        synchronized (lockUpdateTask) {
            // update again if have errors
            if (buildTask.isHasErrors()) {
                scheduleUpdate.compareAndSet(false, true);
            }
            buildTask = null;
        }

        logger.trace("Updating task is finished");
    }

    @Override
    public void scheduleForExecute(final List<Command> commands, final boolean reorderCommands) {
        if (commands.isEmpty()) {
            return;
        }

        final List<? extends Command> queue;
        if (reorderCommands) {
            queue = new ArrayList<Command>(commands);
            Collections.sort(queue, CommandsComparator.getInstance());
        } else {
            queue = commands;
        }

        commandsQueue.addAll(queue);
    }

    @Override
    public void scheduleForExecute(final List<Command> commands) {
        scheduleForExecute(commands, false);
    }

    @Override
    public void scheduleForExecute(final Command command) {
        commandsQueue.add(command);
    }

    @Override
    public void start() {
        initDB();

        // matrix updating

        updateThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (Thread.currentThread().isAlive()) {
                    if (scheduleUpdate.compareAndSet(true, false)) {
                        try {
                            executeUpdateTask();
                        } catch (Exception e) {
                            logger.error("Error while update distance matrices", e);
                        }
                    }

                    try {
                        logger.trace("Wait new changes...");
                        synchronized (scheduleUpdate) {
                            scheduleUpdate.wait(UPDATE_EXECUTION_DELAY); // nothing to do and we wait
                        }
                    } catch (InterruptedException ignore) {
                        break;
                    }
                }
            }
        });

        try {
            updateThread.setPriority(Thread.MIN_PRIORITY);
        } catch (SecurityException e) {
            logger.warn("Can't change priority for thread of matrix updating ", e);
        }

        updateThread.start();

        // commands consumer

        commandsThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    executeCommands();
                } catch (InterruptedException ignore) {
                    //
                }
            }
        });

        try {
            commandsThread.setPriority(Thread.NORM_PRIORITY);
        } catch (SecurityException e) {
            logger.warn("Can't change priority for commands thread", e);
        }

        commandsThread.start();
    }

    private void initDB() {
        this.db = null;
        final GisDataSource gisDataSource = GisDataSources.get("route-cache");
        try {
            this.db = new DistanceMatrixDAOImpl(gisDataSource);
        } catch (IllegalArgumentException e) {
            logger.warn("Data source for DistanceMatrixService not defined in configuration files, all requests will be ignored");
        } catch (NamingException e) {
            logger.error("Error while lookup data source by JNDI name " + gisDataSource, e);
        } catch (SQLException e) {
            logger.error("Error while get schema information for data source " + gisDataSource, e);
        }
    }

    @Override
    public void stop() {
        try {
            eventsProducer.close();
        } catch (Exception e) {
            logger.warn("Can't remove events subscribers", e);
        }

        try {
            if (updateThread != null) {
                updateThread.interrupt();
                updateThread = null;
            }
        } catch (Exception e) {
            logger.warn("Can't interrupt thread of matrix updating", e);
        }

        try {
            if (commandsThread != null) {
                commandsThread.interrupt();
                commandsThread = null;
            }
        } catch (Exception e) {
            logger.warn("Can't interrupt commands thread", e);
        }
    }

    private final static class MatrixBuildTaskGroup {
        private final long groupId;
        private final long matrixId;
        private final long priority;

        private MatrixBuildTaskGroup(final long groupId, final long matrixId, final long priority) {
            this.groupId = groupId;
            this.matrixId = matrixId;
            this.priority = priority;
        }

        public long getGroupId() {
            return groupId;
        }

        public long getMatrixId() {
            return matrixId;
        }

        public long getPriority() {
            return priority;
        }

        @Override
        public boolean equals(final Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;

            final MatrixBuildTaskGroup that = (MatrixBuildTaskGroup) o;

            return groupId == that.groupId && matrixId == that.matrixId;
        }

        @Override
        public int hashCode() {
            int result = (int) (groupId ^ (groupId >>> 32));
            result = 31 * result + (int) (matrixId ^ (matrixId >>> 32));
            return result;
        }
    }
}
